package com.amigo.online.provider.compare.quartzjob;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;

import com.amigo.online.provider.common.ConstantValue;
import com.amigo.online.provider.common.compare.AbstractBasicCompareService;
import com.amigo.online.provider.common.entity.VideoMongoInfo;
import com.amigo.online.provider.common.entity.VideoMysqlInfo;
import com.amigo.online.provider.common.entity.VideoSolrInfo;
import com.amigo.online.provider.compare.mapper.VideoComparisonMapper;
import com.amigo.online.provider.compare.service.VideoComparisonService;
import com.amigo.online.provider.kafka.common.KafkaConstantValue;
import com.amigo.online.provider.kafka.model.CompareFailMessage;
import com.amigo.online.provider.sync.handle.service.DataHandler;

public class AbstractVideoComparisonJob extends AbstractCompareJob {

    @Autowired
    ExecutorService executorService;

    @Autowired
    VideoComparisonMapper videoComparisonMapper;

    @Autowired
    VideoComparisonService videoComparisonService;

    @Autowired
    @Qualifier("videoMongoDataHandler")
    DataHandler videoMongoDataHandler;

    @Autowired
    @Qualifier("videoSolrDataHandler")
    DataHandler videoSolrDataHandler;

    protected void compareVideo(List<VideoMongoInfo> videoMongoList, List<VideoSolrInfo> videoSolrList,
                                List<VideoMysqlInfo> videoMysqlList) {
        Map<Long, VideoMongoInfo> mongoMap = videoMongoList.stream()
                .collect(Collectors.toMap(VideoMongoInfo::getId, videoMongoInfo -> videoMongoInfo));
        Map<Long, VideoSolrInfo> solrMap = videoSolrList.stream()
                .collect(Collectors.toMap(VideoSolrInfo::getId, videoSolrInfo -> videoSolrInfo));
        for (VideoMysqlInfo temp : videoMysqlList) {
            VideoMongoInfo videoMongoInfo = mongoMap.get(temp.getId());

            if (videoMongoInfo == null) {
                CompareFailMessage compareFailMessage = new CompareFailMessage(ConstantValue.TBL_INSERT,
                        ConstantValue.VIDEO_TBL, ConstantValue.SOURCE_MONGO, "", temp.getId());
                kafkaMessageSender.failLogSend(KafkaConstantValue.COMPARSION_MONGO_VIDEO_FAIL, compareFailMessage);
                executorService.execute(() -> {
                    videoMongoDataHandler.insertObject(temp);
                });
            } else {
                Map<String, Object> mongoWrongMap = new HashMap<>(16);
                AbstractBasicCompareService.compare(temp, videoMongoInfo, mongoWrongMap);
                if (!mongoWrongMap.isEmpty()) {
                    executorService.execute(() -> {
                        videoMongoDataHandler.updateInfoByIdThroughHashMap(temp.getId(), mongoWrongMap);
                    });
                }
            }


            VideoSolrInfo videoSolrInfo = solrMap.get(temp.getId());
            Map<String, Object> solrWrongMap = new HashMap<>(32);
            Map<String, Object> mongoWrongMap = new HashMap<>(32);
            if (videoMongoInfo != null) {
                AbstractBasicCompareService.compare(temp, videoMongoInfo, mongoWrongMap);
                if (!mongoWrongMap.isEmpty()) {
                    CompareFailMessage compareFailMessage = new CompareFailMessage(ConstantValue.TBL_UPDATE,
                            ConstantValue.VIDEO_TBL, ConstantValue.SOURCE_MONGO, mongoWrongMap, temp.getId());
                    kafkaMessageSender.failLogSend(KafkaConstantValue.COMPARSION_MONGO_VIDEO_FAIL, compareFailMessage);
                }
            } else {
                CompareFailMessage compareFailMessage = new CompareFailMessage(ConstantValue.TBL_INSERT,
                        ConstantValue.VIDEO_TBL, ConstantValue.SOURCE_MONGO, "", temp.getId());
                kafkaMessageSender.failLogSend(KafkaConstantValue.COMPARSION_MONGO_VIDEO_FAIL, compareFailMessage);
            }
            if (videoSolrInfo != null) {
                AbstractBasicCompareService.compare(temp, videoSolrInfo, solrWrongMap);
                if (!mongoWrongMap.isEmpty()) {
                    CompareFailMessage compareFailMessage = new CompareFailMessage(ConstantValue.TBL_UPDATE,
                            ConstantValue.VIDEO_TBL, ConstantValue.SOURCE_SOLR, mongoWrongMap, temp.getId());
                    kafkaMessageSender.failLogSend(KafkaConstantValue.COMPARSION_SOLR_VIDEO_FAIL, compareFailMessage);
                }
            } else {
                CompareFailMessage compareFailMessage = new CompareFailMessage(ConstantValue.TBL_INSERT,
                        ConstantValue.VIDEO_TBL, ConstantValue.SOURCE_SOLR, "", temp.getId());
                kafkaMessageSender.failLogSend(KafkaConstantValue.COMPARSION_SOLR_VIDEO_FAIL, compareFailMessage);

            }
        }
    }

    /*@Override
    public void execute(JobExecutionContext context) throws JobExecutionException {

        List<VideoMongoInfo> videoMongoList = new ArrayList<>();
        List<VideoSolrInfo> videoSolrList = new ArrayList<>();
        List<VideoMysqlInfo> videoMysqlList = new ArrayList<>();
        context.getJobDetail();
        // mysql 和mongo 数据比对
        System.out.println("Mysql-------to----------mongo");

        int maxId = videoComparisonService.getMaxId();
        // 异步操作，把查询的数据塞入到全局变量中
        int index = maxId / 1000 + 1;

        List<FutureTask<List<VideoMongoInfo>>> mongofutureList = new ArrayList<>(maxId);
        List<FutureTask<List<VideoMysqlInfo>>> mysqlfutureList = new ArrayList<>(maxId);
        List<FutureTask<List<VideoSolrInfo>>> solrfutureList = new ArrayList<>(maxId);
        for (int i = 1; i < index; i++) {
            int startId = i + 1000 * (i - 1);
            int endId = startId + 999;
            // 批量查询，并将结果值存在hashMap中
            MongoVideoTask mongoVideoTask = new MongoVideoTask(startId, endId);
            FutureTask<List<VideoMongoInfo>> mongofuture = new FutureTask<>(mongoVideoTask);
            executorService.submit(mongofuture);
            mongofutureList.add(mongofuture);

            MysqlVideoTask mysqlVideoTask = new MysqlVideoTask(startId, endId);
            FutureTask<List<VideoMysqlInfo>> mysqlfuture = new FutureTask<>(mysqlVideoTask);
            executorService.submit(mysqlfuture);
            mysqlfutureList.add(mysqlfuture);

            SolrVideoTask solrVideoTask = new SolrVideoTask(startId, endId);
            FutureTask<List<VideoSolrInfo>> solrfuture = new FutureTask<>(solrVideoTask);
            executorService.submit(solrfuture);
            solrfutureList.add(solrfuture);

        }
        executorService.shutdown();
        try {
            for (FutureTask<List<VideoMongoInfo>> temp : mongofutureList) {
                List<VideoMongoInfo> partVideoMongoList = temp.get();
                videoMongoList.addAll(partVideoMongoList);
            }
            for (FutureTask<List<VideoMysqlInfo>> temp : mysqlfutureList) {
                List<VideoMysqlInfo> partVideoMysqlList = temp.get();
                videoMysqlList.addAll(partVideoMysqlList);
            }
            for (FutureTask<List<VideoSolrInfo>> temp : solrfutureList) {
                List<VideoSolrInfo> partVideoMongoList = temp.get();
                videoSolrList.addAll(partVideoMongoList);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        compare(videoMongoList, videoSolrList, videoMysqlList);
    }
*/
    private void compare(List<VideoMongoInfo> videoMongoList, List<VideoSolrInfo> videoSolrList,
                         List<VideoMysqlInfo> videoMysqlList) {
        Map<Long, VideoMongoInfo> mongoMap = videoMongoList.stream()
                .collect(Collectors.toMap(VideoMongoInfo::getId, videoMongoInfo -> videoMongoInfo));
        Map<Long, VideoSolrInfo> solrMap = videoSolrList.stream()
                .collect(Collectors.toMap(VideoSolrInfo::getId, videoSolrInfo -> videoSolrInfo));
        for (VideoMysqlInfo temp : videoMysqlList) {
            VideoMongoInfo videoMongoInfo = mongoMap.get(temp.getId());
            VideoSolrInfo videoSolrInfo = solrMap.get(temp.getId());
            Map<String, Object> solrWrongMap = new HashMap<>(32);
            Map<String, Object> mongoWrongMap = new HashMap<>(32);
            AbstractBasicCompareService.compare(temp, videoMongoInfo, mongoWrongMap);
            AbstractBasicCompareService.compare(temp, videoSolrInfo, solrWrongMap);

            videoMongoDataHandler.updateInfoByIdThroughHashMap(temp.getId(), mongoWrongMap);

            videoSolrDataHandler.updateInfoByIdThroughHashMap(temp.getId(), solrWrongMap);

        }
    }

}
